package net.bwie.vehicle.dws.function1;
import net.bwie.vehicle.dws.bean1.VehicleData;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * 每3秒采样一次的处理函数
 * 用于车辆数据监控，确保数据按固定频率输出
 */
public class VehicleMonitoringFunction extends KeyedProcessFunction<String, VehicleData, VehicleData> {
// 状态变量：存储上次发射数据的时间
private transient ValueState<Long> lastEmitTimeState;
// 状态变量：存储最新的车辆数据
private transient ValueState<VehicleData> lastDataState;

/**
 * 初始化操作，包括状态变量的设置
 * @param parameters 配置参数
 */
@Override
public void open(Configuration parameters) {
    try {
        // 初始化状态变量
        lastEmitTimeState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("lastEmitTime", Types.LONG));
        lastDataState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("lastData", Types.POJO(VehicleData.class)));
    } catch (Exception e) {
        throw e;
    }
}

/**
 * 处理数据流中的每个元素
 * @param data 当前车辆数据
 * @param ctx 上下文，用于访问函数的运行时上下文
 * @param out 用于输出数据到下一个操作
 * @throws Exception 抛出异常以处理错误
 */
@Override
public void processElement(
        VehicleData data,
        Context ctx,
        Collector<VehicleData> out
) throws Exception {
    try {
        // 更新最新数据
        lastDataState.update(data);

        // 检查是否需要发射数据
        Long lastEmitTime = lastEmitTimeState.value();
        long currentTime = ctx.timestamp();  // 改用处理时间

        // 如果是第一次检查或自上次发射以来已超过3秒，则发射数据并更新发射时间
        if (lastEmitTime == null || currentTime - lastEmitTime >= 3000) {
            out.collect(data);
            lastEmitTimeState.update(currentTime);

            // 清除旧定时器防止内存泄漏
            ctx.timerService().deleteEventTimeTimer(currentTime - 3000);
            // 注册新的定时器
            scheduleNextTimer(ctx, currentTime);
        }
    } catch (Exception e) {
        throw e;
    }
}

/**
 * 定时器触发时执行的操作
 * @param timestamp 定时器触发的时间戳
 * @param ctx 上下文，用于访问函数的运行时上下文
 * @param out 用于输出数据到下一个操作
 * @throws Exception 抛出异常以处理错误
 */
@Override
public void onTimer(
        long timestamp,
        OnTimerContext ctx,
        Collector<VehicleData> out
) throws Exception {
    try {
        // 定时器触发时发射最新数据
        VehicleData latestData = lastDataState.value();
        if (latestData != null) {
            out.collect(latestData);
            // 注册下一个3秒的定时器
            scheduleNextTimer(ctx, timestamp);
        }
    } catch (Exception e) {
        throw e;
    }
}

/**
 * 注册下一个定时器
 * 提取公共方法避免重复代码
 */
private void scheduleNextTimer(Context ctx, long currentTime) {
    ctx.timerService().registerEventTimeTimer(currentTime + 3000);
}

/**
 * 清理操作，包括清除状态变量
 */
@Override
public void close() {
    try {
        // 清理状态资源
        if (lastEmitTimeState != null) {
            lastEmitTimeState.clear();
        }
        if (lastDataState != null) {
            lastDataState.clear();
        }
    } catch (Exception e) {
        throw e;
    }
}

}
